45. Python自定义函数

本章概要

  • 学习材料:现金流序列、折现率及函数参数合同。
  • 本章任务:运行 lst-function-definitionlst-early-returnlst-google-docstring-check,定义可验证自定义函数。
  • 完成后你将得到:函数签名、返回值、异常分支和docstring检查。
  • 自我检查:手算一组DCF并测试空输入/非法利率;递归无基线或参数失配时先检查原因。
  • 拓展练习:把函数拓展应用到另一人民币现金流。

为什么要学习自定义函数?

  • 抽象化:将复杂实现隐藏在简洁接口之后
  • 模块化:大问题分解为小函数,单一职责
  • 复用性:一次编写,多处调用,减少出错
  • 可读性:函数名就是最好的文档

函数在金融分析中的意义

  • 策略封装:交易策略参数化配置
  • 指标计算:MA、MACD、RSI 等封装为函数
  • 风险管理:VaR、最大回撤等函数化
  • 回测框架:整个流程就是一系列函数的组合

函数定义的基本语法

使用 def 关键字定义函数:

def 函数名(参数列表):
    '''文档字符串'''
    函数体
    return 返回值
  • def:定义函数的关键字
  • 函数名:用 snake_case 命名,动词开头
  • 文档字符串:说明函数用途、参数和返回值
  • return:返回计算结果

函数定义与调用示例

Listing 1: 函数定义与调用的完整示例
展开实现代码
def greet(name):
    """
    问候函数:向指定名字的人问好

    参数:
        name (str): 要问候的人名

    返回:
        str: 包含问候语的字符串
    """
    return f'您好, {name}!'

# 调用函数
# 函数名后跟括号,括号内传入实际参数
message = greet('李四')

# 打印返回值
print(message)  # 输出: 您好, 李四!

# 可以在表达式中直接使用函数返回值
print(greet('张三'))  # 输出: 您好, 张三!
您好, 李四!
您好, 张三!

函数调用机制

Listing 2: 函数调用机制的详细演示
展开实现代码
def demonstrate_call_process(x, y):
    """
    演示函数调用的完整过程
    """
    print(f'接收到参数: x={x}, y={y}')

    # 执行函数逻辑
    result = x + y

    print(f'计算结果: {result}')

    # 返回结果
    return result

# 函数调用过程分解
# 1. 创建栈帧(Stack Frame):为函数调用分配内存空间
# 2. 参数传递:将实参值绑定到形参名
# 3. 执行函数体:按顺序执行函数内语句
# 4. 返回值:将结果返回给调用者
# 5. 销毁栈帧:释放函数调用占用的内存

a, b = 10, 20
print('准备调用函数...')

# 函数调用:程序跳转到函数体执行
value = demonstrate_call_process(a, b)

print(f'函数返回值: {value}')
print('函数调用结束,继续执行主程序')
准备调用函数...
接收到参数: x=10, y=20
计算结果: 30
函数返回值: 30
函数调用结束,继续执行主程序

占位函数只用于讲解开发语法

pass... 只演示“先搭框架”的语法,不是练习答案。业务分析写出物必须以有输入、有输出、可验证的完整函数替换占位实现。

Listing 3: 空函数定义的占位符写法
展开代码
def placeholder_function():
    """
    这是一个占位函数,稍后实现

    使用pass语句避免语法错误
    """
    pass  # pass是Python的空操作语句

# 调用空函数不会产生任何效果
placeholder_function()

# 另一种常见的占位写法:使用省略号(...)
def another_placeholder():
    """另一个占位函数"""
    ...  # 省略号在Python中也是有效的表达式

another_placeholder()  # 不会报错

位置参数

最基础的参数类型,按照定义时的顺序传递:

Listing 4: 位置参数的基本用法
展开实现代码
def calculate_profit(revenue, cost):
    """
    计算利润的函数

    参数:
        revenue (float): 营业收入
        cost (float): 营业成本

    返回:
        float: 利润额
    """
    profit = revenue - cost
    return profit

# 调用时必须按照定义的顺序传递参数
profit1 = calculate_profit(1000, 600)
print(f'利润1: {profit1}')  # 输出: 利润1: 400

# 如果顺序错误,会得到错误结果
profit2 = calculate_profit(600, 1000)
print(f'利润2: {profit2}')  # 输出: 利润2: -400 (错误!)
利润1: 400
利润2: -400

关键字参数

通过参数名传递,不受顺序限制,代码更清晰:

Listing 5: 关键字参数的使用
展开代码
def calculate_profit(revenue, cost):
    """计算利润"""
    return revenue - cost

# 使用关键字参数,顺序可以任意
profit1 = calculate_profit(revenue=1000, cost=600)
profit2 = calculate_profit(cost=600, revenue=1000)

print(f'利润1: {profit1}')  # 输出: 利润1: 400
print(f'利润2: {profit2}')  # 输出: 利润2: 400 (正确!)

# 关键字参数提高了代码的可读性
# 尤其当函数有多个参数时,不容易混淆
利润1: 400
利润2: 400

默认参数

为参数提供默认值,使函数调用更灵活:

Listing 6: 默认参数的定义和使用
展开实现代码
def calculate_tax(salary, rate=0.05, insurance_rate=0.02):
    """
    计算税后工资

    参数:
        salary (float): 税前工资
        rate (float): 税率,默认5%
        insurance_rate (float): 社保扣除比例,默认2%

    返回:
        float: 税后工资
    """
    # 计算税金
    tax = salary * rate

    # 计算社保扣除
    insurance = salary * insurance_rate

    # 计算实发工资
    net_salary = salary - tax - insurance

    return net_salary

# 情况1:只提供必填参数
net1 = calculate_tax(5000)
print(f'税后工资1(默认税率): {net1:.2f}元')

# 情况2:覆盖默认税率
net2 = calculate_tax(5000, rate=0.10)
print(f'税后工资2(10%税率): {net2:.2f}元')

# 情况3:使用关键字参数指定特定参数
net3 = calculate_tax(5000, insurance_rate=0.05)
print(f'税后工资3(5%社保): {net3:.2f}元')

# 情况4:所有参数都指定
net4 = calculate_tax(5000, rate=0.08, insurance_rate=0.03)
print(f'税后工资4(全部指定): {net4:.2f}元')
税后工资1(默认税率): 4650.00元
税后工资2(10%税率): 4400.00元
税后工资3(5%社保): 4500.00元
税后工资4(全部指定): 4450.00元

可变默认参数陷阱

使用列表等可变对象作为默认参数时要特别小心:

Listing 7: 可变默认参数的陷阱演示
展开实现代码
# 错误示例:使用列表作为默认参数
def append_item(item, items=[]):  # 危险!
    """向列表添加项目"""
    items.append(item)
    return items

# 第一次调用
result1 = append_item('A')
print(f'第一次: {result1}')  # 输出: ['A']

# 第二次调用
result2 = append_item('B')
print(f'第二次: {result2}')  # 输出: ['A', 'B'] 意外!

# 问题:默认列表在函数定义时创建,所有调用共享同一个列表

# 正确做法:使用None作为默认值
def append_item_correct(item, items=None):
    """向列表添加项目(正确版本)"""
    if items is None:  # 每次调用都创建新列表
        items = []
    items.append(item)
    return items

# 测试正确版本
result3 = append_item_correct('A')
print(f'正确第一次: {result3}')  # 输出: ['A']

result4 = append_item_correct('B')
print(f'正确第二次: {result4}')  # 输出: ['B'] 符合预期!
第一次: ['A']
第二次: ['A', 'B']
正确第一次: ['A']
正确第二次: ['B']

可变位置参数 *args

接收任意数量的位置参数:

Listing 8: 可变位置参数的使用
展开实现代码
def sum_all(*args):
    """
    计算所有传入数字的和

    参数:
        *args: 可变数量的位置参数

    返回:
        float/int: 所有参数的和
    """
    print(f'接收到 {len(args)} 个参数: {args}')

    # args是一个元组,包含所有位置参数
    total = sum(args)

    return total

# 调用示例
result1 = sum_all(1, 2, 3)
print(f'和1: {result1}')  # 输出: 和1: 6

result2 = sum_all(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
print(f'和2: {result2}')  # 输出: 和2: 55

# 也可以传递列表或元组,使用*解包
numbers = [10, 20, 30, 40]
result3 = sum_all(*numbers)  # *将列表解包为独立参数
print(f'和3: {result3}')  # 输出: 和3: 100
接收到 3 个参数: (1, 2, 3)
和1: 6
接收到 10 个参数: (1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
和2: 55
接收到 4 个参数: (10, 20, 30, 40)
和3: 100

可变关键字参数 **kwargs

接收任意数量的关键字参数:

Listing 9: 可变关键字参数的使用
展开实现代码
def create_portfolio(**kwargs):
    """
    创建投资组合

    参数:
        **kwargs: 股票代码和持仓数量的键值对

    返回:
        dict: 投资组合字典
    """
    print('构建投资组合...')

    # kwargs是一个字典,包含所有关键字参数
    for symbol, shares in kwargs.items():
        print(f'  {symbol}: {shares}股')

    return kwargs

# 调用示例
portfolio1 = create_portfolio(AAPL=100, MSFT=50, GOOGL=30)
print(f'组合1: {portfolio1}')

# 也可以传递字典,使用**解包
stocks = {'600519.SH': 500, '000858.SZ': 1000}
portfolio2 = create_portfolio(**stocks)  # **将字典解包为关键字参数
print(f'组合2: {portfolio2}')
构建投资组合...
  AAPL: 100股
  MSFT: 50股
  GOOGL: 30股
组合1: {'AAPL': 100, 'MSFT': 50, 'GOOGL': 30}
构建投资组合...
  600519.SH: 500股
  000858.SZ: 1000股
组合2: {'600519.SH': 500, '000858.SZ': 1000}

参数组合规则

多种参数类型同时使用时,必须遵循顺序:

Listing 10: 函数参数的正确顺序
展开实现代码
def complex_function(a, b, c=10, *args, **kwargs):
    """
    演示各种参数的组合使用

    参数顺序规则:
    1. 位置参数 (a, b)
    2. 默认参数 (c=10)
    3. 可变位置参数 (*args)
    4. 可变关键字参数 (**kwargs)
    """
    print(f'a={a}, b={b}, c={c}')
    print(f'args: {args}')
    print(f'kwargs: {kwargs}')

# 调用示例
complex_function(1, 2, 3, 4, 5, x=100, y=200)

# 输出:
# a=1, b=2, c=3
# args: (4, 5)
# kwargs: {'x': 100, 'y': 200}
a=1, b=2, c=3
args: (4, 5)
kwargs: {'x': 100, 'y': 200}

单一返回值

Listing 11: 单一返回值函数
展开实现代码
def calculate_return(initial, final):
    """
    计算简单收益率

    参数:
        initial (float): 初始价格
        final (float): 最终价格

    返回:
        float: 收益率(小数形式)
    """
    return (final - initial) / initial

# 使用示例
initial_price = 100
final_price = 115

return_rate = calculate_return(initial_price, final_price)
print(f'收益率: {return_rate:.2%}')  # 输出: 收益率: 15.00%
收益率: 15.00%

多返回值

Python 函数可以返回多个值(实际是元组):

以下 7 个价格点代表一年内 6 个等长收益区间,因此设 periods_per_year=6;年无风险利率先按复利转换到同一期间频率,再计算年化 Sharpe。

Listing 12: 函数返回多个值
展开实现代码
import math, statistics  # 导入数学与统计工具以统一收益风险指标的年化频率

def calculate_return_metrics(prices, periods_per_year, annual_risk_free_rate):  # 定义带频率约定的收益风险函数
    """返回总收益率、路径最大回撤、年化收益率、年化波动率和年化 Sharpe。"""
    if len(prices) < 3 or periods_per_year <= 0 or annual_risk_free_rate <= -1 or any(price <= 0 for price in prices):  # 核验价格、区间数与无风险利率口径
        raise ValueError('至少需要3个正价格,且频率为正、年无风险利率大于-100%')  # 阻止无效输入生成误导性风险指标
    return_intervals = len(prices) - 1  # 用相邻价格之间的区间数而非观察点数年化
    period_returns = [prices[index] / prices[index - 1] - 1 for index in range(1, len(prices))]  # 计算同频区间收益率
    total_return = prices[-1] / prices[0] - 1  # 计算整个观察窗口的累计收益率
    annualized_return = (1 + total_return) ** (periods_per_year / return_intervals) - 1  # 按实际收益区间数年化
    running_peak = prices[0]  # 初始化时序路径的运行峰值
    max_drawdown = 0.0  # 初始化峰值到后续谷值的最大回撤
    for current_price in prices:  # 按时间顺序检查每个价格点
        running_peak = max(running_peak, current_price)  # 仅允许历史及当前价格更新运行峰值
        current_drawdown = (running_peak - current_price) / running_peak  # 计算当前点相对先前峰值的回撤
        max_drawdown = max(max_drawdown, current_drawdown)  # 保留路径上最大的峰谷损失比例
    period_volatility = statistics.stdev(period_returns)  # 用区间收益率而非价格水平计算波动率
    annualized_volatility = period_volatility * math.sqrt(periods_per_year)  # 按同一频率年化收益率波动
    period_risk_free_rate = (1 + annual_risk_free_rate) ** (1 / periods_per_year) - 1  # 将年无风险利率复利转换为区间利率
    sharpe_ratio = None if period_volatility == 0 else (statistics.mean(period_returns) - period_risk_free_rate) / period_volatility * math.sqrt(periods_per_year)  # 用同频超额收益与波动率计算年化 Sharpe
    return total_return, max_drawdown, annualized_return, annualized_volatility, sharpe_ratio  # 返回五项口径一致的业务指标

prices = [100, 105, 102, 108, 112, 110, 115]  # 设置一年内六个等长区间的课堂价格路径
metrics = calculate_return_metrics(prices, periods_per_year=6, annual_risk_free_rate=0.03)  # 按六期频率和3%年无风险利率计算指标
assert math.isclose(metrics[1], 3 / 105, rel_tol=1e-12) and math.isclose(metrics[2], metrics[0], rel_tol=1e-12)  # 验证路径回撤与六区间年化不变量
assert calculate_return_metrics([90, 100, 110], 2, 0.03)[1] == 0.0  # 验证先低后高的路径不会制造回撤
print(f'总收益率: {metrics[0]:.2%}\n最大回撤: {metrics[1]:.2%}\n年化收益率: {metrics[2]:.2%}\n年化波动率: {metrics[3]:.2%}\n年化Sharpe: {metrics[4]:.2f}')  # 输出同一频率合同下的五项指标
总收益率: 15.00%
最大回撤: 2.86%
年化收益率: 15.00%
年化波动率: 9.19%
年化Sharpe: 1.25

提前返回(Early Return)

函数可以在多个地方返回,处理边界条件:

Listing 13: 提前返回的使用技巧
展开实现代码
def calculate_yearly_return(prices, periods_per_year):  # 定义带明确期间频率的年化收益函数
    """按相邻价格形成的收益区间数计算年化收益率,无效输入返回 None。"""
    if len(prices) < 2:  # 价格观察不足时无法形成收益区间
        print('错误:至少需要2个价格点')  # 向学习者说明缺少可比区间
        return None  # 在无可比期间时提前停止
    if periods_per_year <= 0 or any(price <= 0 for price in prices):  # 核验频率与全部价格均可用于复利计算
        print('错误:频率必须为正且所有价格必须为正数')  # 向学习者说明复利口径无效
        return None  # 在复利口径无效时提前停止
    return_intervals = len(prices) - 1  # 将观察点数量转换为相邻收益区间数
    yearly_return = (prices[-1] / prices[0]) ** (periods_per_year / return_intervals) - 1  # 按区间数和每年频率年化
    return yearly_return  # 返回频率口径明确的年化收益率

test1 = calculate_yearly_return([100, 105, 110], periods_per_year=2)  # 用一年两个等长区间验证10%年收益
test2 = calculate_yearly_return([100], periods_per_year=2)  # 验证观察点不足的提前返回
test3 = calculate_yearly_return([0, 100], periods_per_year=1)  # 验证非正价格的提前返回
assert math.isclose(test1, 0.10, rel_tol=1e-12) and test2 is None and test3 is None  # 验证区间年化和两类输入守卫
print(f'测试1(正常): {test1:.2%}\n测试2(数据不足): {test2}\n测试3(价格为0): {test3}')  # 汇总展示三条执行路径
错误:至少需要2个价格点
错误:频率必须为正且所有价格必须为正数
测试1(正常): 10.00%
测试2(数据不足): None
测试3(价格为0): None

无返回值与 None

没有 return 语句的函数返回 None

Listing 14: 无返回值函数
展开实现代码
def log_message(message):
    """记录日志消息"""
    print(f'[LOG] {message}')
    # 没有return语句,返回None

result = log_message('系统启动')
print(f'返回值: {result}')  # 输出: 返回值: None
print(f'返回值类型: {type(result)}')  # 输出: <class 'NoneType'>

def explicit_none():
    """显式返回None"""
    print('执行完毕')
    return None  # 显式返回None

result2 = explicit_none()
print(f'返回值2: {result2}')  # 输出: 返回值2: None
[LOG] 系统启动
返回值: None
返回值类型: <class 'NoneType'>
执行完毕
返回值2: None

递归函数:什么是递归?

递归是函数调用自身的编程技巧。正确的递归必须包含:

  • 基准情况(Base Case):递归终止的条件
  • 递归情况(Recursive Case):将问题分解为更小的子问题

递归示例:计算阶乘

Listing 15: 递归函数计算阶乘
展开实现代码
def factorial(n):
    """
    计算阶乘:n! = n × (n-1) × ... × 2 × 1

    参数:
        n (int): 非负整数

    返回:
        int: n的阶乘

    递归关系:
        - 0! = 1 (基准情况)
        - n! = n × (n-1)! (递归情况)
    """
    # 基准情况:递归的终止条件
    if n <= 1:
        return 1

    # 递归情况:将问题分解
    # n! = n × (n-1)!
    return n * factorial(n - 1)

# 使用示例
print(f'0! = {factorial(0)}')  # 输出: 0! = 1
print(f'5! = {factorial(5)}')  # 输出: 5! = 120
print(f'10! = {factorial(10)}')  # 输出: 10! = 3628800

# 递归执行过程演示(以factorial(3)为例)
# factorial(3)
# → 3 * factorial(2)
#     → 2 * factorial(1)
#         → 1 (基准情况)
#     = 2 * 1 = 2
# = 3 * 2 = 6
0! = 1
5! = 120
10! = 3628800

递归在金融中的应用:折现现金流

估值时点为 \(t=0\),列表第一个元素是 \(t=1\) 年末现金流;因此每笔现金流都至少折现一期。

Listing 16: 递归计算现值
展开实现代码
import math  # 导入数学工具以核验平价债券的现值不变量

def present_value(cash_flows, discount_rate, period=1):  # 从估值日后的第一期开始递归折现
    """
    使用递归计算现金流的现值

    参数:
        cash_flows (list): 现金流列表
        discount_rate (float): 折现率
        period (int): 当前年末期数,默认从1开始(内部使用)

    返回:
        float: 所有现金流的现值

    递归关系:
        PV = CF₁/(1+r) + CF₂/(1+r)² + CF₃/(1+r)³ + ...
    """
    if period > len(cash_flows):  # 超过最后一个年末现金流时结束递归
        return 0.0  # 返回零以终止剩余现值累加
    current_cash_flow = cash_flows[period - 1]  # 将第一个列表元素映射到第一年末
    discount_factor = (1 + discount_rate) ** period  # 按年末期数计算折现因子
    current_present_value = current_cash_flow / discount_factor  # 折现当前年末现金流
    remaining_present_value = present_value(cash_flows, discount_rate, period + 1)  # 递归折现后续年末现金流
    return current_present_value + remaining_present_value  # 汇总当前与后续现金流现值

cash_flows = [50, 50, 1050]  # 设置票息率等于折现率的三年期平价债券现金流
rate = 0.05  # 设置与5%票息率一致的年折现率
pv = present_value(cash_flows, rate)  # 从第一年末开始计算债券现值
assert math.isclose(pv, 1000.0, rel_tol=1e-12)  # 验证票息率等于折现率时债券价格等于面值
print(f'现值: {pv:.2f}元')  # 输出与递归讲解一致的1000元现值

# 递归调用过程:
# present_value([50, 50, 1050], 0.05, 1)
# → 50/(1.05)¹ + present_value([50, 50, 1050], 0.05, 2)
#     → 50/(1.05)² + present_value([50, 50, 1050], 0.05, 3)
#         → 1050/(1.05)³ + present_value([50, 50, 1050], 0.05, 4)
#             → 0 (基准情况)
# = 1000.0
现值: 1000.00元

递归的深度限制

Listing 17: 递归深度限制演示
展开实现代码
import sys

# 查看默认递归深度限制
print(f'递归深度限制: {sys.getrecursionlimit()}')

# 尝试超过限制会导致错误
def deep_recursion(n):
    """深度递归演示"""
    if n <= 0:
        return 0
    return n + deep_recursion(n - 1)

# 这会导致 RecursionError
try:
    result = deep_recursion(2000)
except RecursionError as e:
    print(f'错误:递归深度超限 - {e}')

# 解决方法:使用循环代替递归
def iterative_sum(n):
    """循环版本,无深度限制"""
    total = 0
    for i in range(n + 1):
        total += i
    return total

result = iterative_sum(2000)
print(f'循环版本结果: {result}')
递归深度限制: 3000
循环版本结果: 2001000

变量作用域:LEGB 规则

Python 使用 LEGB 规则查找变量:

层级 名称 含义
L Local 局部作用域(函数内部)
E Enclosing 嵌套函数的外层函数作用域
G Global 全局作用域(模块级别)
B Built-in 内置作用域(Python 内置模块)

变量作用域示例

Listing 18: 变量作用域演示
展开实现代码
# 全局变量(G层)
x = 10  # 全局变量x
y = 100  # 全局变量y

def outer_function():
    """外层函数(E层)"""

    # 外层函数的局部变量
    y = 20  # 遮蔽全局变量y

    def inner_function():
        """内层函数(L层)"""

        # 内层函数的局部变量
        z = 5  # 局部变量z

        # 访问各层变量
        print(f'内层局部变量 z: {z}')
        print(f'外层变量 y: {y}')
        print(f'全局变量 x: {x}')

    # 调用内层函数
    inner_function()

# 调用外层函数
outer_function()

# 在全局作用域访问x和y
print(f'全局变量 x: {x}')
print(f'全局变量 y: {y}')

# 尝试访问局部变量会报错
try:
    print(z)
except NameError as e:
    print(f'错误:局部变量在外部不可访问 - {e}')
内层局部变量 z: 5
外层变量 y: 20
全局变量 x: 10
全局变量 x: 10
全局变量 y: 100
错误:局部变量在外部不可访问 - name 'z' is not defined

global 关键字

在函数内部修改全局变量:

Listing 19: 使用global修改全局变量
展开实现代码
# 全局变量
counter = 0
total_sum = 0

def increment_counter():
    """增加计数器"""
    global counter  # 声明使用全局变量
    counter += 1  # 修改全局变量
    print(f'计数器: {counter}')

def add_to_total(value):
    """累加到总和"""
    global total_sum  # 声明使用全局变量
    total_sum += value
    print(f'累加后总和: {total_sum}')

# 测试
print(f'初始counter: {counter}')
increment_counter()  # counter变为1
increment_counter()  # counter变为2
print(f'最终counter: {counter}')

print(f'初始total_sum: {total_sum}')
add_to_total(100)  # total_sum变为100
add_to_total(200)  # total_sum变为300
print(f'最终total_sum: {total_sum}')
初始counter: 0
计数器: 1
计数器: 2
最终counter: 2
初始total_sum: 0
累加后总和: 100
累加后总和: 300
最终total_sum: 300

nonlocal 关键字

嵌套函数中修改外层函数的变量:

Listing 20: 使用nonlocal修改外层变量
展开实现代码
def make_accumulator():
    """累加器工厂函数"""

    # 外层函数的局部变量
    total = 0

    def accumulator(value):
        """内层累加函数"""
        nonlocal total  # 声明使用外层变量
        total += value
        return total

    return accumulator  # 返回内层函数

# 创建累加器
acc1 = make_accumulator()
acc2 = make_accumulator()

# 每个累加器有独立的状态
print(f'acc1累加10: {acc1(10)}')  # 输出: 10
print(f'acc1累加20: {acc1(20)}')  # 输出: 30
print(f'acc2累加100: {acc2(100)}')  # 输出: 100 (独立的total)
acc1累加10: 10
acc1累加20: 30
acc2累加100: 100

lambda 表达式:匿名函数

lambda 是创建匿名函数的简洁方式:

Listing 21: lambda表达式的应用
展开实现代码
# lambda语法:lambda 参数列表: 表达式
# 特点:单个表达式,自动返回结果

# 示例1:简单平方函数
square = lambda x: x**2
print(f'5的平方: {square(5)}')  # 输出: 5的平方: 25

# 示例2:等价于以下普通函数
def square_normal(x):
    return x**2

print(f'6的平方: {square_normal(6)}')  # 输出: 6的平方: 36

# 示例3:多参数lambda
add = lambda x, y: x + y
print(f'3 + 5 = {add(3, 5)}')  # 输出: 3 + 5 = 8

# 示例4:lambda在金融中的应用 - 计算投资净值
net_value = lambda prices, shares: sum(p * s for p, s in zip(prices, shares))

prices = [10.5, 20.3, 15.8]
shares = [100, 200, 150]

value = net_value(prices, shares)
print(f'投资净值: {value:.2f}元')
5的平方: 25
6的平方: 36
3 + 5 = 8
投资净值: 7480.00元

lambda 的典型应用场景

Listing 22: lambda表达式的典型应用
展开实现代码
# 场景1:与map()结合 - 对序列中每个元素应用函数
numbers = [1, 2, 3, 4, 5]
squared = list(map(lambda x: x**2, numbers))
print(f'平方列表: {squared}')  # 输出: [1, 4, 9, 16, 25]

# 场景2:与filter()结合 - 筛选符合条件的元素
stocks = [
    ('AAPL', 150),
    ('MSFT', 280),
    ('GOOGL', 2700)
]

# 筛选价格大于200的股票
expensive = list(filter(lambda s: s[1] > 200, stocks))
print(f'高价股票: {expensive}')  # 输出: [('MSFT', 280), ('GOOGL', 2700)]

# 场景3:与sorted()结合 - 自定义排序
portfolio = [
    {'symbol': 'AAPL', 'shares': 100},
    {'symbol': 'MSFT', 'shares': 50},
    {'symbol': 'GOOGL', 'shares': 200}
]

# 按持股数量排序
sorted_by_shares = sorted(portfolio, key=lambda x: x['shares'])
print(f'按持股排序:')
for item in sorted_by_shares:
    print(f"  {item['symbol']}: {item['shares']}股")

# 场景4:与max()和min()结合 - 基于复杂条件找极值
# 找出收益率最高的股票
stock_returns = [('AAPL', 0.15), ('MSFT', 0.20), ('GOOGL', 0.12)]
best_stock = max(stock_returns, key=lambda x: x[1])
print(f'最佳股票: {best_stock[0]}, 收益率: {best_stock[1]:.2%}')
平方列表: [1, 4, 9, 16, 25]
高价股票: [('MSFT', 280), ('GOOGL', 2700)]
按持股排序:
  MSFT: 50股
  AAPL: 100股
  GOOGL: 200股
最佳股票: MSFT, 收益率: 20.00%

lambda 的局限性

lambda 只适合简短逻辑,复杂情况用普通函数:

Listing 23: lambda的局限性演示
展开实现代码
# 不好的lambda示例:过于复杂
# calculate_tax_bad = lambda salary: salary * 0.05 if salary <= 5000 else salary * 0.10

# 更好的做法:使用普通函数
def calculate_tax(salary):
    """
    计算个人所得税

    参数:
        salary (float): 税前工资

    返回:
        float: 应缴税额
    """
    if salary <= 5000:
        return salary * 0.05
    else:
        return salary * 0.10

# 使用
print(f'税额(3000): {calculate_tax(3000):.2f}元')
print(f'税额(8000): {calculate_tax(8000):.2f}元')
税额(3000): 150.00元
税额(8000): 800.00元

高阶函数:map()

map() 对序列中每个元素应用函数:

Listing 24: map()函数的使用
展开实现代码
def calculate_return(prices):
    """计算价格序列的收益率"""
    returns = []
    for i in range(1, len(prices)):
        ret = (prices[i] - prices[i-1]) / prices[i-1]
        returns.append(ret)
    return returns

# 使用map处理多个股票
stocks = {
    'AAPL': [150, 152, 148, 155],
    'MSFT': [280, 285, 282, 290],
    'GOOGL': [2700, 2720, 2680, 2750]
}

# 对每只股票应用收益率计算
returns_dict = dict(map(
    lambda item: (item[0], calculate_return(item[1])),
    stocks.items()
))

print('各股票收益率:')
for symbol, returns in returns_dict.items():
    print(f'{symbol}: {[f"{r:.2%}" for r in returns]}')
各股票收益率:
AAPL: ['1.33%', '-2.63%', '4.73%']
MSFT: ['1.79%', '-1.05%', '2.84%']
GOOGL: ['0.74%', '-1.47%', '2.61%']

高阶函数:filter()

filter() 筛选符合条件的元素:

Listing 25: filter()函数的使用
展开代码
stocks = [
    {'symbol': 'AAPL', 'price': 150, 'pe': 25},
    {'symbol': 'MSFT', 'price': 280, 'pe': 30},
    {'symbol': 'GOOGL', 'price': 2700, 'pe': 20},
    {'symbol': 'T业务规定', 'price': 800, 'pe': 100}
]

# 筛选PE < 30的股票
value_stocks = list(filter(lambda s: s['pe'] < 30, stocks))

print('价值股票(PE < 30):')
for stock in value_stocks:
    print(f"  {stock['symbol']}: PE={stock['pe']}")
价值股票(PE < 30):
  AAPL: PE=25
  GOOGL: PE=20

高阶函数:reduce()

reduce() 对序列做累积计算:

Listing 26: reduce()函数的使用
展开实现代码
from functools import reduce

def calculate_compound_return(returns):
    """
    计算复合收益率

    参数:
        returns (list): 期收益率列表

    返回:
        float: 复合收益率
    """
    # 复合收益率 = (1+r₁) × (1+r₂) × ... × (1+rₙ) - 1
    # 使用reduce累积计算
    compound = reduce(
        lambda acc, r: acc * (1 + r),  # 累积乘法
        returns,
        1.0  # 初始值
    )
    return compound - 1

# 示例:4个季度的收益率
quarterly_returns = [0.05, 0.03, 0.08, 0.02]

annual_return = calculate_compound_return(quarterly_returns)
print(f'年复合收益率: {annual_return:.2%}')
年复合收益率: 19.14%

函数文档:Google 风格 Docstring

Listing 27: Google风格文档字符串
展开完整参考实现
def calculate_sharpe_ratio(returns, periods_per_year=12, annual_risk_free_rate=0.03):  # 定义带期间频率与年无风险利率合同的夏普函数
    """
    计算同频收益率的年化夏普比率。

    参数:
        returns (list[float]): 同频小数收益率,至少两个有限值。
        periods_per_year (int): 每年收益区间数。
        annual_risk_free_rate (float): 年有效无风险利率。

    返回:
        float | None: 年化比率;零波动时返回None。

    异常:
        ValueError: 样本、频率或利率合同不满足时。
    """
    import math  # 导入数学工具以按期间频率年化夏普比率
    import statistics  # 导入统计工具以计算同频收益率波动
    if len(returns) < 2 or periods_per_year <= 0 or annual_risk_free_rate <= -1:  # 核验样本、频率与无风险利率合同
        raise ValueError('至少需要2个收益率,且频率为正、年无风险利率大于-100%')  # 阻止不可比口径进入计算
    period_volatility = statistics.stdev(returns)  # 计算同频区间收益率的样本波动率
    if period_volatility == 0:  # 同频收益没有波动时无法定义夏普比率
        return None  # 避免以零波动率作为分母
    period_risk_free_rate = (1 + annual_risk_free_rate) ** (1 / periods_per_year) - 1  # 将年无风险利率复利转换到同频期间
    sharpe = (statistics.mean(returns) - period_risk_free_rate) / period_volatility * math.sqrt(periods_per_year)  # 年化同频超额收益风险比
    return sharpe  # 返回频率合同一致的年化夏普比率
Listing 28
展开函数合同核对
test_returns = [0.05, 0.03, 0.08, -0.02, 0.06]  # 设置五个月度课堂收益率
sharpe_ratio = calculate_sharpe_ratio(test_returns, periods_per_year=12, annual_risk_free_rate=0.03)  # 按月度频率与3%年无风险利率计算
assert math.isclose(sharpe_ratio, 3.414509694460191, rel_tol=1e-12)  # 锁定同频转换后的年化夏普不变量
print(f'年化夏普比率: {sharpe_ratio:.2f}')  # 输出月度收益合同下的年化结果
年化夏普比率: 3.41

函数设计最佳实践

  • 单一职责:每个函数只做一件事
  • 描述性命名:使用动词短语,如 calculate_sharpe_ratio()
  • 参数数量:不超过 5 个位置参数
  • 函数长度:理想不超过 20 行
  • 避免副作用:纯函数更易于测试和推理

总结

知识点 核心要点
函数定义 def + 函数名 + 参数 + return
参数类型 位置、关键字、默认、*args**kwargs
返回值 单返回值、多返回值(元组)、None
递归 基准情况 + 递归情况,注意深度限制
作用域 LEGB 规则,global/nonlocal
lambda 匿名函数,适合简短的一次性逻辑

运行前预测|平台任务1:计算利率平均值

  • 输入预测:运行前先写出 dataShiborLiborHibor 的业务含义、数据类型或取值范围,并判断哪一个输入最可能改变结果。
  • 结果预测:不展开答案,先预测将得到“3月11日到29日期间3个月期Shibor平均值”相关结果;同时写出方向、数量级或表格/图形结构。
  • 完成要求:能独立说明本任务从输入到“平台任务1:计算利率平均值”结果的关键步骤,原样录入平台代码并得到可核对的运行结果。

⭐ 平台任务1:计算利率平均值

展开完整代码(投影默认折叠)
# ⚠️ 平台原始代码 - 请原样输入至教学平台(注释除外),平台才会判定答案正确
import pandas as pd  # 导入Pandas数据分析库

# 从Excel文件读取数据存入data
data = pd.read_excel(r"https://huoran.oss-cn-shenzhen.aliyuncs.com/20220811/xlsx/1557658167503511552.xlsx")

data.columns =["日期","Shibor","Libor","Hibor"]  # 定义列表data.columns

Shibor = []  # 定义列表Shibor

for i in data["Shibor"]:  # 遍历data["Shibor"]中的每个i

    Shibor.append(i)  # 将Shibor利率数据添加到列表

Libor = []  # 定义列表Libor

for i in data["Libor"]:  # 遍历data["Libor"]中的每个i

    Libor.append(i)  # 将美元Libor利率数据添加到列表

Hibor = []  # 定义列表Hibor

for i in data["Hibor"]:  # 遍历data["Hibor"]中的每个i
    
    Hibor.append(i)  # 将港元Hibor利率数据添加到列表

f_mean = lambda x: sum(x)/len(x) #用lambda函数定义并且参数x以列表的数据结构输入

Shibor_mean = f_mean(x=Shibor)  #计算Shibor的平均值

Libor_mean = f_mean(x=Libor)   #计算美元Libor的平均值

Hibor_mean = f_mean(x=Hibor)   #计算美元Hibor的平均值

print("3月11日到29日期间3个月期Shibor平均值",round(Shibor_mean,6))  # 输出3月11日到29日期间3个月期Shibor平均值

print("3月11日到29日期间3个月期美元Libor平均值",round(Libor_mean,6))  # 输出3月11日到29日期间3个月期美元Libor平均值

print("3月11日到29日期间3个月期Hibor平均值",round(Hibor_mean,6))  # 输出3月11日到29日期间3个月期Hibor平均值

任务复盘|平台任务1:计算利率平均值

运行后核对:核对 dataShiborLiborHibor 是否按预测参与运算,实际输出是否与预测一致;若不一致,先检查类型、单位、索引/字段和运算顺序。

拓展练习:只改变一个关键输入或业务场景,先预测输出如何变化,再运行验证并解释变化原因。

运行前预测|平台任务2:计算利率波动率

  • 输入预测:运行前先写出 dataShiborLiborHibor 的业务含义、数据类型或取值范围,并判断哪一个输入最可能改变结果。
  • 结果预测:不展开答案,先预测将得到“3月11日到29日期间3个月期Shibor波动率”相关结果;同时写出方向、数量级或表格/图形结构。
  • 完成要求:能独立说明本任务从输入到“平台任务2:计算利率波动率”结果的关键步骤,原样录入平台代码并得到可核对的运行结果。

⭐ 平台任务2:计算利率波动率

展开完整代码(投影默认折叠)
# ⚠️ 平台原始代码 - 请原样输入至教学平台(注释除外),平台才会判定答案正确
import pandas as pd  # 导入Pandas数据分析库

# 从Excel文件读取数据存入data
data = pd.read_excel(r"https://huoran.oss-cn-shenzhen.aliyuncs.com/20220811/xlsx/1557658167503511552.xlsx")

data.columns =["日期","Shibor","Libor","Hibor"]  # 定义列表data.columns

Shibor = []  # 定义列表Shibor

for i in data["Shibor"]:  # 遍历data["Shibor"]中的每个i

    Shibor.append(i)  # 将Shibor利率数据添加到列表

Libor = []  # 定义列表Libor

for i in data["Libor"]:  # 遍历data["Libor"]中的每个i

    Libor.append(i)  # 将美元Libor利率数据添加到列表

Hibor = []  # 定义列表Hibor

for i in data["Hibor"]:  # 遍历data["Hibor"]中的每个i

    Hibor.append(i)  # 将港元Hibor利率数据添加到列表

def f_sigma(x):  # 定义函数f_sigma

  '''通过Python定义一个计算变量波动率的函数

  x:代表变量的样本值,可以用列表的数据结构输入'''

  n = len(x)  # 获取数据长度

  u_mean = sum(x)/n #计算变量样本值的均值

  z = []       #生成一个空列表

  for t in range(n):  # 遍历range(n)中的每个t

        z.append((x[t]-u_mean)**2)  # 将计算结果累加到z列表

  return (sum(z)/(n-1))**0.5    #计算波动率      

Shibor_sigma = f_sigma(x=Shibor) #计算Shibor波动率

Libor_sigma = f_sigma(x=Libor) #计算Libor波动率

Hibor_sigma = f_sigma(x=Hibor) #计算Hibor波动率

print("3月11日到29日期间3个月期Shibor波动率",round(Shibor_sigma,6))  # 输出3月11日到29日期间3个月期Shibor波动率

print("3月11日到29日期间3个月期没有Libor波动率",round(Libor_sigma,6))  # 输出3月11日到29日期间3个月期没有Libor波动率

print("3月11日到29日期间3个月期Hibor波动率",round(Hibor_sigma,6))  # 输出3月11日到29日期间3个月期Hibor波动率

任务复盘|平台任务2:计算利率波动率

运行后核对:核对 dataShiborLiborHibor 是否按预测参与运算,实际输出是否与预测一致;若不一致,先检查类型、单位、索引/字段和运算顺序。

拓展练习:只改变一个关键输入或业务场景,先预测输出如何变化,再运行验证并解释变化原因。

运行前预测|平台任务3:利率比较

  • 输入预测:运行前先写出 Shibor_meanLibor_meanHibor_meanShibor_sigma 的业务含义、数据类型或取值范围,并判断哪一个输入最可能改变结果。
  • 结果预测:不展开答案,先预测将得到Hibor_sigma 的结果;同时写出方向、数量级或表格/图形结构。
  • 完成要求:能独立说明本任务从输入到“平台任务3:利率比较”结果的关键步骤,原样录入平台代码并得到可核对的运行结果。

⭐ 平台任务3:利率比较

展开完整代码(投影默认折叠)
# ⚠️ 平台原始代码 - 请原样输入至教学平台(注释除外),平台才会判定答案正确
Shibor_mean = 0.028065  # 设置期数/数量为0

Libor_mean = 0.026077  # 设置期数/数量为0

Hibor_mean = 0.017619  # 设置期数/数量为0

Shibor_sigma = 0.000286  # 设置数组长度参数为0

Libor_sigma = 0.000111  # 设置数组长度参数为0

Hibor_sigma = 0.000702  # 设置数组长度参数为0

print(Shibor_mean > Libor_mean) #判断Shibor的平均值是否大于美元Libor的平均值

print(Libor_mean > Hibor_mean)  #判断美元Libor的平均值是否大于Hibor的平均值

print(Shibor_sigma > Libor_sigma) #判断Shibor的波动率是否大于Libor的波动率

print(Libor_sigma > Hibor_sigma) #判断美云Libor的波动率是否大于Hibor的波动率

print(Shibor_sigma > Hibor_sigma) #判断Shibor的波动率是否大于Hibor的波动率

任务复盘|平台任务3:利率比较

运行后核对:核对 Shibor_meanLibor_meanHibor_meanShibor_sigma 是否按预测参与运算,实际输出是否与预测一致;若不一致,先检查类型、单位、索引/字段和运算顺序。

拓展练习:只改变一个关键输入或业务场景,先预测输出如何变化,再运行验证并解释变化原因。

随堂练习

  • 问题 1|需要准备哪些数据?:现金流序列、折现率及函数参数合同。
  • 问题 2|需要完成哪些操作?:运行 lst-function-definitionlst-early-returnlst-google-docstring-check,定义可验证自定义函数。
  • 问题 3|应得到哪些结果?:函数签名、返回值、异常分支和docstring检查。
  • 问题 4|怎样确认结果可靠?:手算一组DCF并测试空输入/非法利率;递归无基线或参数失配时先检查原因。
  • 问题 5|换一个情境,怎样继续应用?:把函数拓展应用到另一人民币现金流。
  • 作答提示:请依次写清所用数据、分析过程、所得结果、核对方法和拓展思考。课程所需数据见前言中的下载入口;教学平台固定题按页面说明完成。

教师参考解答|答案与说明 1

  • 所用数据与字段:现金流序列 cash_flows 与折现率 discount_rate;折现率必须大于 −1,现金流不得为空。

教师参考解答|代码 1

展开代码(代码区可独立滚动)
def calculate_dcf(cash_flows,discount_rate):  # 定义现金流折现合同
    """返回从t=0开始的现金流现值;拒绝空输入和不合法利率。"""  # 说明时间口径与失败分支
    if not cash_flows: raise ValueError('cash_flows不能为空')  # 空序列无法形成估值依据
    if discount_rate<=-1: raise ValueError('discount_rate必须大于-1')  # 避免折现因子为零或符号失真
    return sum(cash_flow/(1+discount_rate)**period for period,cash_flow in enumerate(cash_flows))  # 按t=0口径逐期折现
two_period_value=calculate_dcf([-100,110],.10)  # 构造可手算的两期现金流
assert abs(two_period_value-0)<1e-10  # 用-100+110/1.1=0另行核对
print(round(two_period_value,2))  # 输出两位小数参考结果

教师参考解答|答案与说明 2

  • 解释答案:签名同时声明输入与返回;docstring 固定 t=0 时间口径;空输入和 r≤−1 通过同一接口请先检查数据后再继续。
  • 拓展应用答案:替换为另一组人民币现金流并保持时间口径;先手算两期,再比较函数输出,不把 DCF 正值直接等同于投资建议。
  • 参考结果:正常示例得到 0.00;空列表和利率为 −1 的情况应分别报错。
  • 常见错误:混淆 t=0/t=1、遗漏空输入、允许 r≤−1